package join;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.JobConf;

import global.ArgumentParam;
import global.ArgumentType;
import global.BaseDriver;
import global.GlobalConsts;
import histogram.Histogram;

public class Join extends BaseDriver {
	static int binH0=1000;
	static int binH1=2000;
	static int bottomValueH0 = 0;
	static int bottomValueH1 = 0;
	Map<Integer,ArrayList<Integer>> mapH0;
	Map<Integer,ArrayList<Integer>> mapH1;
	ArrayList<HistogramBin> actionResults;
	static int k = 100000;
	int gk;
	int threshold;
	int currentBinId = 0;
	int sum = 0;
	int maxBinIdH1 = Integer.MAX_VALUE;

	public Join(String[] args) {
		super(args, JoinConsts.jobName);
	}
	
	public static void main(String[] args) {
		new Join(args);
	}
	
	@Override
	protected ArrayList<ArgumentParam> fillParamsList() {
		ArrayList<ArgumentParam> res = new ArrayList<ArgumentParam>(4);
		
		ArgumentParam arg;
		
		arg = new ArgumentParam(GlobalConsts.paramNameInputPath, GlobalConsts.defaultRelationsPath, ArgumentType.INPUT_FOLDER, false);
		res.add(arg);
		
		arg = new ArgumentParam(GlobalConsts.paramNameOutputPath, GlobalConsts.defaultOutputPath, ArgumentType.OUTPUT_FOLDER, false);
		res.add(arg);
		
		arg = new ArgumentParam(GlobalConsts.paramNameBinSizeR0, GlobalConsts.defaultBinSizeR0 + "", ArgumentType.GENERAL_INT, true);
		res.add(arg);
		
		arg = new ArgumentParam(GlobalConsts.paramNameBinSizeR1, GlobalConsts.defaultBinSizeR1 + "", ArgumentType.GENERAL_INT, true);
		res.add(arg);
		
		return res;
	}

	@Override
	protected JobConf createJobConf() {
		JobConf conf = new JobConf(Join.class);
		return conf;
	}
	
	@Override
	protected void doPreProcess() {
		String lineH0;
		String lineH1;
		mapH0 = new HashMap<Integer, ArrayList<Integer>>();
		mapH1 = new HashMap<Integer, ArrayList<Integer>>();
		actionResults = new ArrayList<HistogramBin>();
		ArrayList<Integer> tempListH0 = new ArrayList<Integer>();
		ArrayList<Integer> tempListH1 = new ArrayList<Integer>();
		boolean stopReading = false;
		FileSystem hdfs = null;
		Path pathH0, pathH1;
		try {
			hdfs =  FileSystem.get(getJobConf());
		} catch (IOException e1) {
			e1.printStackTrace();
		}
		try {
			pathH0 = new Path(GlobalConsts.defaultHistogramPath + "/" + GlobalConsts.fileNameHistogram0);
			pathH1 = new Path(GlobalConsts.defaultHistogramPath + "/" + GlobalConsts.fileNameHistogram1);
			BufferedReader inStreamH0 = new BufferedReader(new InputStreamReader(hdfs.open(pathH0)));
			BufferedReader inStreamH1 = new BufferedReader(new InputStreamReader(hdfs.open(pathH1)));
			//printBinId();
			while ((lineH0 = inStreamH0.readLine()) != null) {
				if (currentBinId != Integer.valueOf(lineH0.split(" ")[1])){
					mapH0.put(currentBinId, tempListH0);
					tempListH0 = new ArrayList<Integer>();
					stopReading = false;
					while (!stopReading && maxBinIdH1>=currentBinId) {
						lineH1 = inStreamH1.readLine();
						if(lineH1 != null){
							if (currentBinId != Integer.valueOf(lineH1.split(" ")[1])){
								mapH1.put(currentBinId, tempListH1);
								tempListH1 = new ArrayList<Integer>();
							
								stopReading = true;
							}
							tempListH1.add(Integer.valueOf(lineH1.split("	")[1]));
						}
						else{
							maxBinIdH1 = currentBinId;
							stopReading = true;
							mapH1.put(currentBinId, tempListH1);
							tempListH1 = new ArrayList<Integer>();
						}
					}
					calculate();
					//printSum();
					if((sum > k) && (gk <= 0)){
						gk = actionResults.get(0).getTopBin();
					}
					threshold = calculateThreshold();
					//printGk();
					//printThreshold();
					if(isEnd()) System.exit(0);
					currentBinId = Integer.valueOf(lineH0.split(" ")[1]);
					//printBinId();
				}
				tempListH0.add(Integer.valueOf(lineH0.split("	")[1]));
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		finally {
			try {
				hdfs.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
	private int applyFunction(int h0Bin, int h1Bin){
		return h0Bin + h1Bin;
	}
	private void calculate(){
		for(int i = 0; i < mapH0.get(currentBinId).size(); i++){
			int upTo;
			if(currentBinId < maxBinIdH1) upTo = currentBinId;
			else upTo = maxBinIdH1;
			for(int j = 0; j <= upTo; j++){
				actionResults.add(new HistogramBin(applyFunction(currentBinId*binH0 + bottomValueH0, j*binH1 + bottomValueH1), 
						applyFunction(((currentBinId*binH0) + binH0 -1 + bottomValueH0), ((j*binH1) + binH1 -1 + bottomValueH1)),
				mapH0.get(currentBinId).get(i) * mapH1.get(j).get(i)));
				sum += mapH0.get(currentBinId).get(i) * mapH1.get(j).get(i);
			}
		}
		//printData(true);
		Collections.sort(actionResults);
		//printData(false);
	}
	private int calculateThreshold(){
		if( (applyFunction(((currentBinId * binH0) + binH0 - 1 + bottomValueH0) , bottomValueH1)) < (applyFunction(((currentBinId * binH1) + binH1 - 1 + bottomValueH1) , bottomValueH0))){
			return applyFunction(((currentBinId * binH0) + binH0 - 1 + bottomValueH0) , bottomValueH1);
		}
		else{
			return applyFunction(((currentBinId * binH1) + binH1 - 1 + bottomValueH1) , bottomValueH0);
		}
	}
	private boolean isEnd(){
		if (gk > 0){
			if(threshold > gk){
				printBounds();
				return true;
			}
		}
		return false;
	}
	private void printBounds(){
		System.out.println();
		System.out.println("##############################");
		System.out.println("topBins= " + ((currentBinId * binH0) + binH0 - 1 + bottomValueH0) + " και " + ((currentBinId * binH1) + binH1 - 1 + bottomValueH1));
		System.out.println();
	}
}
